package com.atguigu.tingshu.common.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class KafkaService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaService.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;


    public Boolean sendMsg(String topics, String value) {
        //  发送消息：
        CompletableFuture completableFuture = kafkaTemplate.send(topics, value);
        //  需要判断消息是否发送成功
        //        completableFuture.whenComplete((o, throwable) -> {
        //            if (throwable == null) {
        //                logger.info("消息发送成功");
        //            } else {
        //                logger.error("消息发送失败");
        //            }
        //        });
        completableFuture.thenAcceptAsync(o -> {
            logger.info("消息发送成功");
        }).exceptionally(throwable -> {
            //  如何保证 kafka 消息正确投递?  重试机制!  error消息记录表; insert into error_msg values( topics value ); sendMsg() 或 sendEmail();
            logger.error("消息发送失败");
            return false;
        });
        return true;
    }
}
